package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.util.ClickHouseUtil;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author caodan
 * @version 1.0
 * @date 2022-10-10 14:47
 * 流量域版本-渠道-地区-访客类别粒度页面浏览各窗口轻度聚合  案例涉及多流合并操作 为此选用union算子
 */
public class DwsTrafficVcChArIsNewPageViewWindow {

    public static void main(String[] args) throws Exception {

        // 创建流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        // 设置检查点相关信息

        // 从kafka中获取相关数据
        // 从 dwd_traffic_page_log 获取数据
        FlinkKafkaConsumer<String> dwdTrafficPageLog = KafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log", "");
        // 从 dwd_traffic_unique_visitor_detail 获取数据
        FlinkKafkaConsumer<String> dwdTrafficUniqueVisitorDetail = KafkaUtil.getFlinkKafkaConsumer("dwd_traffic_unique_visitor_detail", "");
        // 从 dwd_traffic_user_jump_detail 获取数据
        FlinkKafkaConsumer<String> dwdTrafficUserJumpDetail = KafkaUtil.getFlinkKafkaConsumer("dwd_traffic_user_jump_detail", "");
        DataStreamSource<String> pageLogDS = env.addSource(dwdTrafficPageLog);
        DataStreamSource<String> uniqueVisitDS = env.addSource(dwdTrafficUniqueVisitorDetail);
        DataStreamSource<String> userJumpDetailDS = env.addSource(dwdTrafficUserJumpDetail);

//        pageLogDS.print(">>>");
//        uniqueVisitDS.print("@@@");
//        userJumpDetailDS.print("$$$");

        // 对读取的数据做类型转换 JSONStr-> 实体类对象
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewBeanByPageLog = pageLogDS.map((MapFunction<String, TrafficPageViewBean>) value -> {
            JSONObject jsonObj = JSON.parseObject(value);
            JSONObject common = jsonObj.getJSONObject("common");
            JSONObject page = jsonObj.getJSONObject("page");

            // 获取 ts
            Long ts = jsonObj.getLong("ts");

            // 获取维度信息
            String vc = common.getString("vc");
            String ch = common.getString("ch");
            String ar = common.getString("ar");
            String isNew = common.getString("is_new");

            // 获取页面访问时长
            Long duringTime = page.getLong("during_time");

            // 定义变量接受其它度量值
            long svCt = 0L;

            // 判断本页面是否开启了一个新的会话
            String lastPageId = page.getString("last_page_id");
            if (lastPageId == null || lastPageId.length() == 0) {
                svCt = 1L;
            }

            // 封装为实体类
            return new TrafficPageViewBean(
                    "",
                    "",
                    vc,
                    ch,
                    ar,
                    isNew,
                    0L,
                    svCt,
                    1L,
                    duringTime,
                    0L,
                    ts
            );
        });
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewBeanByUniqueVisit = uniqueVisitDS.map((MapFunction<String, TrafficPageViewBean>) value -> {
            JSONObject jsonObj = JSON.parseObject(value);
            JSONObject common = jsonObj.getJSONObject("common");
            Long ts = jsonObj.getLong("ts");

            // 获取维度信息
            String vc = common.getString("vc");
            String ch = common.getString("ch");
            String ar = common.getString("ar");
            String isNew = common.getString("is_new");

            // 封装为实体类
            return new TrafficPageViewBean(
                    "",
                    "",
                    vc,
                    ch,
                    ar,
                    isNew,
                    1L,
                    0L,
                    0L,
                    0L,
                    0L,
                    ts
            );
        });
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewBeanByJumpDetail = userJumpDetailDS.map((MapFunction<String, TrafficPageViewBean>) value -> {
            JSONObject jsonObj = JSONObject.parseObject(value);
            JSONObject common = jsonObj.getJSONObject("common");
            Long ts = jsonObj.getLong("ts") + 10 * 1000L;

            // 获取维度信息
            String vc = common.getString("vc");
            String ch = common.getString("ch");
            String ar = common.getString("ar");
            String isNew = common.getString("is_new");

            // 封装为实体类
            return new TrafficPageViewBean(
                    "",
                    "",
                    vc,
                    ch,
                    ar,
                    isNew,
                    0L,
                    0L,
                    0L,
                    0L,
                    1L,
                    ts
            );
        });
        // 将三条流数据进行合并操作
        DataStream<TrafficPageViewBean> dataStream = trafficPageViewBeanByPageLog.union(trafficPageViewBeanByJumpDetail, trafficPageViewBeanByUniqueVisit);

        //插入水位线
        SingleOutputStreamOperator<TrafficPageViewBean> watermarks = dataStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((SerializableTimestampAssigner<TrafficPageViewBean>) (element, recordTimestamp) -> element.getTs())
        );

        // 对流中数据进行分组，开窗，聚合操作
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedStream = watermarks.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                return Tuple4.of(
                        trafficPageViewBean.getVc(),
                        trafficPageViewBean.getCh(),
                        trafficPageViewBean.getAr(),
                        trafficPageViewBean.getIsNew());
            }
        });
        keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .allowedLateness(Time.seconds(10))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        value1.setUjCt(value1.getUjCt() + value2.getUjCt());
                        return value1;
                    }
                }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        for (TrafficPageViewBean element : input) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    } // 将结果写入到 clickhouse 中
                })
                .addSink(ClickHouseUtil.getJdbcSink("insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
